package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCancelDetail {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //生产环境中设置为kafka主题的分区数

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

//id
//order_id
//sku_id
//sku_name
//order_price
//sku_num
//create_time
//source_type_id
//source_type_name
//source_id
//split_total_amount
//split_activity_amount
//split_coupon_amount
//consignee
//consignee_tel
//total_amount
//order_status
//user_id
//payment_way
//delivery_address
//order_comment
//out_trade_no
//trade_body
//operate_time
//expire_time
//process_status
//tracking_no
//parent_order_id
//province_id
//activity_reduce_amount
//coupon_reduce_amount
//original_total_amount
//feight_fee
//feight_fee_reduce
//refundable_time
//order_detail_activity_id
//activity_id
//activity_rule_id
//order_detail_coupon_id
//coupon_id
//coupon_use_id
//type
//old
        //读取kafka订单预处理表
        tenv.executeSql("CREATE TABLE dwd_order_pre (\n" +
                "  `id` STRING,\n" +
                "  `order_id` STRING,\n" +
                "  `sku_id` STRING,\n" +
                "  `sku_name` STRING,\n" +
                "  `order_price` STRING,\n" +
                "  `sku_num` STRING,\n" +
                "  `create_time` STRING,\n" +
                "  `source_type_id` STRING,\n" +
                "  `source_type_name` STRING,\n" +
                "  `source_id` STRING,\n" +
                "  `split_total_amount` STRING,\n" +
                "  `split_activity_amount` STRING,\n" +
                "  `split_coupon_amount` STRING,\n" +
                "  `consignee` STRING,\n" +
                "  `consignee_tel` STRING,\n" +
                "  `total_amount` STRING,\n" +
                "  `order_status` STRING,\n" +
                "  `user_id` STRING,\n" +
                "  `payment_way` STRING,\n" +
                "  `delivery_address` STRING,\n" +
                "  `order_comment` STRING,\n" +
                "  `out_trade_no` STRING,\n" +
                "  `trade_body` STRING,\n" +
                "  `operate_time` STRING,\n" +
                "  `expire_time` STRING,\n" +
                "  `process_status` STRING,\n" +
                "  `tracking_no` STRING,\n" +
                "  `parent_order_id` STRING,\n" +
                "  `province_id` STRING,\n" +
                "  `activity_reduce_amount` STRING,\n" +
                "  `coupon_reduce_amount` STRING,\n" +
                "  `original_total_amount` STRING,\n" +
                "  `feight_fee` STRING,\n" +
                "  `feight_fee_reduce` STRING,\n" +
                "  `refundable_time` STRING,\n" +
                "  `order_detail_activity_id` STRING,\n" +
                "  `activity_id` STRING,\n" +
                "  `activity_rule_id` STRING,\n" +
                "  `order_detail_coupon_id` STRING,\n" +
                "  `coupon_id` STRING,\n" +
                "  `coupon_use_id` STRING,\n" +
                "  `type` STRING,\n" +
                "  `old` Map<STRING,STRING>\n" +
                ")"+ MyKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process","cancel_detail"));

        //过滤出取消订单数据
        Table filteredTable = tenv.sqlQuery(""+
                "select " +
                "id, " +
                "order_id, " +
                "user_id, " +
                "sku_id, " +
                "sku_name, " +
                "province_id , " +
                "activity_id, " +
                "activity_rule_id, " +
                "coupon_id, " +
                "operate_time cancel_time, " +
                "source_id, " +
                "source_type_id, " +
                "source_type_name, " +
                "sku_num, " +
                "order_price, " +
                "split_activity_amount, " +
                "split_coupon_amount, " +
                "split_total_amount " +
                "from dwd_order_pre " +
                "where `type`='update' " +
                "and `old`['order_status'] is not null " +
                "and order_status='1003'");
        tenv.createTemporaryView("filtered_table",filteredTable);

        //创建kafka 取消订单表
        tenv.executeSql("create table dwd_trade_cancel_detail("+
        "id string,"+
        "order_id string,"+
        "user_id string,"+
        "sku_id string,"+
        "sku_name string,"+
        "province_id string,"+
        "activity_id string,"+
        "activity_rule_id string,"+
        "coupon_id string,"+
        "cancel_time string,"+
        "source_id string,"+
        "source_type_id string,"+
        "source_type_name string,"+
        "sku_num string,"+
        "order_price string,"+
        "split_activity_amount string,"+
        "split_coupon_amount string,"+
        "split_total_amount string"+
        ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cancel_detail"));


        //将数据写出到kafka
        tenv.executeSql("insert into dwd_trade_cancel_detail select * from filtered_table");



    }
}
